package com.dddframework.mq.impl;

import cn.hutool.core.thread.GlobalThreadPool;
import com.dddframework.core.contract.MQEvent;
import com.dddframework.mq.core.MQClient;
import com.dddframework.mq.core.MQFilter;
import com.dddframework.mq.core.MQListener;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.stereotype.Component;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.JedisPubSub;

import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.function.Consumer;
import java.util.function.Supplier;

/**
 * Redis客户端实现
 */
@Slf4j(topic = "### BASE-MQ : redisClient ###")
@Component
public final class RedisClient implements MQClient, DisposableBean {
    private static JedisPool JEDIS_POOL;
    private static List<Supplier> LISTENERS = new ArrayList<>();

    @Override
    public String impl() {
        return "redis";
    }

    @Override
    public Consumer<MQEvent> initProducer() {
        return mqEvent -> {
            String message = serialization().serialize(mqEvent);
            log.info("Publish MQ: {}", message);
            // channel=namespace:topic:tag或namespace:topic或topic
            String channel = String.format("%s%s%s", (config().getNamespace().isEmpty() ? "" : config().getNamespace() + ":"),
                    mqEvent.getTopic(),
                    (mqEvent.getTag() == null || mqEvent.getTag().isEmpty()) ? "" : (":" + mqEvent.getTag()));
            GlobalThreadPool.submit(() -> {
                try {
                    jedis().publish(channel, message);
                } catch (Throwable e) {
                    log.error("Publish MQ to [{}] failed: {}", channel, message, e);
                }
            });
        };
    }

    @Override
    public void initConsumer(MQListener mqListener) throws Exception {
        // channel=namespace:topic:tag或namespace:topic或topic，TODO 暂不支持通配符*和非-
        List<String> channels = new ArrayList<>();
        if (mqListener.getTags() != null && !mqListener.getTags().isEmpty()) {
            Set<String> tags = MQFilter.findIncludes(mqListener.getTags());
            if (!tags.isEmpty()) {
                for (String tag : tags) {
                    channels.add(config().namespace(":") + mqListener.getTopic() + ":" + tag);
                }
            }
        } else {
            channels.add(config().namespace(":") + mqListener.getTopic());
        }
        // Redis原子加锁脚本
        String lockScript = "if redis.call('get', KEYS[1]) == false then redis.call('setex', KEYS[1], tonumber(ARGV[1]), KEYS[1]) return 1 else return 0 end";
        LISTENERS.add(() -> {
            GlobalThreadPool.submit(() -> {
                try (Jedis jedis = jedis()) {
                    jedis.subscribe(new JedisPubSub() {
                        @Override
                        public void onMessage(String channel, String message) {
                            MQEvent mqEvent = mqListener.getDeserialize().apply(message);
                            try {
                                // 按组消费：Redis本身没有消费组的概念，这里我们锁住某条消息的消费资格10秒钟，以实现同消费组（如多实例）只能由一个消费者消费
                                String lockKey = String.format("ChannelGroupLock:%s:%s:%s", channel, mqListener.getGroup(), mqEvent.getMsgId());
                                if ((long) jedis().eval(lockScript, 1, lockKey, "10000") == 1) {
                                    // 本条消息在该消费者组未消费
                                    consume(mqListener, mqEvent);
                                }
                            } catch (Exception e) {
                                log.error("Consume MQ failed: {}", mqEvent, e);
                            }
                        }

                        @Override
                        public void onSubscribe(String channel, int subscribedChannels) {
                            log.info("Subscribed channel: {}", channel);
                        }

                        @Override
                        public void onUnsubscribe(String channel, int subscribedChannels) {
                            log.info("Unsubscribed channel: {}", channel);
                        }
                    }, channels.toArray(new String[0]));
                }
            });
            return null;
        });
    }

    @Override
    public void start() {
        if (!config().isEnable() || !Objects.equals(config().getImpl(), impl())) return;
        LISTENERS.forEach(Supplier::get);
    }

    @Override
    public void destroy() throws Exception {
        if (JEDIS_POOL != null) {
            JEDIS_POOL.close();
        }
    }

    private Jedis jedis() {
        if (JEDIS_POOL == null) {
            String[] hostAndPort = config().getServer().split(":");
            JedisPoolConfig poolConfig = new JedisPoolConfig();
            poolConfig.setMaxTotal(100);
            poolConfig.setMaxIdle(50);
            poolConfig.setMinIdle(10);
            poolConfig.setTestOnBorrow(true);
            if (!config().getPassword().isEmpty()) {
                JEDIS_POOL = new JedisPool(poolConfig, hostAndPort[0], Integer.parseInt(hostAndPort[1]), 5000, config().getPassword());
            } else {
                JEDIS_POOL = new JedisPool(poolConfig, hostAndPort[0], Integer.parseInt(hostAndPort[1]));
            }
        }
        return JEDIS_POOL.getResource();
    }

}